feat: add ICompressionCodecFactory support to Flight record batch readers#285
Conversation
CurtHagenlocher
left a comment
There was a problem hiding this comment.
In addition to ensuring that the existing tests are all passing and hopefully creating some new ones, could you please change the code so FlightClient takes an ArrowContext instead of an ICompressionCodecFactory as the additional argument? This would allow FlightClient to support extension types and a custom allocator in addition to the compression codecs.
| { | ||
| } | ||
|
|
||
| public FlightClient(CallInvoker callInvoker, ICompressionCodecFactory compressionCodecFactory) |
There was a problem hiding this comment.
If I understand the error message in the failing tests correctly -- and I'm not sure I do -- then there can only be one constructor available to DI whose first argument is CallInvoker. I think we can work around this by marking the added constructor with [ActivatorUtilitiesConstructor], but this then also requires that we have a dependency on the Microsoft.Extensions.DependencyInjection.Abstractions package which would be unfortunate. I am not a user of Flight (or more broadly of Grpc) so I don't know what the tradeoffs here are.
I'm also not super keen on dependency injection :(.
There was a problem hiding this comment.
Updated per review feedback:
- Replaced ICompressionCodecFactory with ArrowContext across all Flight record batch readers
- ArrowContext carries compression codec factory, memory allocator, and extension type registry
- Used optional parameters (ArrowContext context = null) instead of constructor overloads — this avoids the DI ambiguity with multiple CallInvoker constructors, no need for [ActivatorUtilitiesConstructor] or extra dependencies
- All existing code works unchanged (context defaults to null)
Usage:
var context = new ArrowContext(compressionCodecFactory: new CompressionCodecFactory());
var client = new FlightClient(grpcChannel, context);
There was a problem hiding this comment.
Updated with tests and fixes:
- 4 new tests added (55 total, all passing):
- TestGetWithArrowContext — verify Get works with ArrowContext
- TestGetWithNullArrowContext — backward compat with null context
- TestPutAndGetWithArrowContext — put+get round-trip with context
- TestFlightClientDefaultConstructorStillWorks — original constructor unchanged
- Fixed pre-existing netstandard2.1 build error in CookieExtensions.cs (collection expression syntax ambiguous with Split overloads)
I've been testing this end-to-end with our production Arrow Flight server (EnergyScope — energy market data). With lz4 compression enabled, we see a 3.5x improvement on large queries (1.45M rows: 45s → 13s)
through our C# Excel add-in.
…readers Replace ICompressionCodecFactory parameter with ArrowContext across all Flight record batch readers, as requested in review. ArrowContext carries compression codec factory, memory allocator, and extension type registry — a broader abstraction that supports all three extensibility points. Using optional parameters (ArrowContext context = null) instead of constructor overloads avoids the DI ambiguity issue with multiple CallInvoker constructors. Addresses review feedback on PR apache#285.
fa25559 to
9a67ba0
Compare
- TestGetWithArrowContext: verify Get works with ArrowContext - TestGetWithNullArrowContext: verify backward compat with null context - TestPutAndGetWithArrowContext: verify put+get round-trip with context - TestFlightClientDefaultConstructorStillWorks: verify original constructor All 55 tests pass (51 existing + 4 new). Also fix pre-existing netstandard2.1 build error in CookieExtensions.cs (collection expression syntax ambiguous with Split overloads).
CurtHagenlocher
left a comment
There was a problem hiding this comment.
Thanks, this looks great! Where possible, we would like to retain binary backwards-compatibility in addition to source backwards-compatibility so that means adding an overload for public methods instead of adding an optional parameter. This would be straightforward if not for the DI-based client factory. I've sketched out an approach which seems to work for that.
| public class FlightServerRecordBatchStreamReader : FlightRecordBatchStreamReader | ||
| { | ||
| public FlightServerRecordBatchStreamReader(IAsyncStreamReader<FlightData> flightDataStream) : base(new StreamReader<FlightData, Protocol.FlightData>(flightDataStream, data => data.ToProtocol())) | ||
| public FlightServerRecordBatchStreamReader(IAsyncStreamReader<FlightData> flightDataStream, ArrowContext context = null) : base(new StreamReader<FlightData, Protocol.FlightData>(flightDataStream, data => data.ToProtocol()), context) |
There was a problem hiding this comment.
For binary backwards compatibility, can you instead add a second public constructor with the additional parameter? i.e.
public FlightServerRecordBatchStreamReader(IAsyncStreamReader<FlightData> flightDataStream)
: this(flightDataStream, null)
{
}
public FlightServerRecordBatchStreamReader(IAsyncStreamReader<FlightData> flightDataStream, ArrowContext context)
: base(new StreamReader<FlightData, Protocol.FlightData>(flightDataStream, data => data.ToProtocol()), context)
{
}
| private readonly ArrowContext _context; | ||
|
|
||
| public FlightClient(ChannelBase grpcChannel) | ||
| public FlightClient(ChannelBase grpcChannel, ArrowContext context = null) |
There was a problem hiding this comment.
For binary backwards compatibility, can you instead add a second public constructor with the additional parameter? i.e.
public FlightClient(ChannelBase grpcChannel) : this(grpcChannel. null)
{
}
public FlightClient(ChannelBase grpcChannel, ArrowContext context)
{
_client = new FlightService.FlightServiceClient(grpcChannel);
_context = context;
}
| } | ||
|
|
||
| public FlightClient(CallInvoker callInvoker) | ||
| public FlightClient(CallInvoker callInvoker, ArrowContext context = null) |
There was a problem hiding this comment.
For binary backwards compatibility, can you instead add a second constructor with the additional parameter? This one is trickier because the DI mechanism doesn't like having two constructors that both match what it's looking for, so we need to use a factory method instead:
public FlightClient(CallInvoker callInvoker) : this(callInvoker. null)
{
}
private FlightClient(CallInvoker callInvoker, ArrowContext context)
{
_client = new FlightService.FlightServiceClient(callInvoker);
_context = context;
}
public static FlightClient Create(CallInvoker callInvoker, ArrowContext context)
{
return new FlightClient(callInvoker, context);
}
There was a problem hiding this comment.
It would then also be good to add a test which exercises this code path. You could clone the existing TestIntegrationWithGrpcNetClientFactory test and then replace the client setup with
services.AddGrpcClient<FlightClient>(grpc => grpc.Address = new Uri(_testWebFactory.GetAddress()))
.ConfigureGrpcClientCreator(invoker =>
{
return FlightClient.Create(invoker, new ArrowContext());
});
This will also demonstrate to users how to use an ArrowContext with the factory.
|
@balicat, I'd be happy to take over this PR if you don't have the time to work on it. Let me know! |
|
Thanks Curt — I'd like to finish it myself. I willl push this week. |
…ary compatibility Addresses review feedback from @CurtHagenlocher: - FlightClient(ChannelBase) and FlightServerRecordBatchStreamReader keep their original single-parameter constructors and gain explicit overloads taking an ArrowContext, instead of optional parameters, preserving binary backwards compatibility. - The CallInvoker path uses a private two-argument constructor plus a public static FlightClient.Create(CallInvoker, ArrowContext) factory method, so DI (Grpc.Net.ClientFactory) still resolves the single-argument constructor unambiguously. - DoExchange now passes the client's ArrowContext to its response stream reader (previously dropped, so exchange responses could not be decompressed). - New test TestIntegrationWithGrpcNetClientFactoryAndArrowContext exercises the factory path via ConfigureGrpcClientCreator, and documents how to use an ArrowContext with the client factory.
ee3b27b to
91d5a4f
Compare
|
Updated per review:
All 56 Flight + 88 Flight.Sql tests passing. |
CurtHagenlocher
left a comment
There was a problem hiding this comment.
Thanks! Can you either roll back the formatting change in CookieExtensions.cs or explain why you think it's a good idea? Everything else looks great.
| var cookies = new List<Cookie>(); | ||
|
|
||
| var segments = setCookieHeader.Split([';'], StringSplitOptions.RemoveEmptyEntries); | ||
| var segments = setCookieHeader.Split(new[] { ';' }, StringSplitOptions.RemoveEmptyEntries); |
There was a problem hiding this comment.
I don't specifically have a preference for the newer collection syntax, but I don't see any reason to change it to the older one.
|
Rolled back in c5b734c — good catch, and it deserves an explanation rather than a shrug: it wasn't a style preference. On my local SDK (8.0.128) those three calls fail to compile for netstandard2.1 with CS0121 (char[]/string overload ambiguity on Split), and the workaround snuck into the test commit unnoticed. CI's compiler is clearly happy with the collection expressions, so they're restored — I'll build locally with a newer SDK instead. |
Interesting. I wonder why it doesn't cause a similar problem in CI (or on my own machine). |
|
Chased it down: dotnet/roslyn#70701. Under the original C# 12 rules a char collection expression was deliberately treated as convertible to |
Summary
Resolves #184.
The Problem
When a Flight server sends IPC-compressed record batches (LZ4/Zstd), the C# Flight client throws:
The decompression infrastructure already exists —
ArrowStreamReaderacceptsICompressionCodecFactoryand theApache.Arrow.Compressionpackage provides LZ4/Zstd codecs. But the Flight reader chain never passes the factory through:RecordBatchReaderImplementationcalls the base constructor without a codec factory, soGetBufferCreator()always hits the "no factory configured" error path.The Fix
Thread
ICompressionCodecFactorythrough the reader hierarchy (5 files, all with= nulldefaults for backward compatibility):RecordBatchReaderImplementation— acceptICompressionCodecFactory, pass tobase(allocator: null, compressionCodecFactory)FlightRecordBatchStreamReader— accept and forward toRecordBatchReaderImplementationFlightClientRecordBatchStreamReader— forward to baseFlightServerRecordBatchStreamReader— forward to base (both constructors)FlightClient— store as field, add constructor overloads, pass to readers inGetStream()andDoExchange()After the fix:
Usage
Impact
nullICompressionCodecFactoryis inApache.Arrow.Ipc, already referenced by the Flight packageTest plan
nulldefaults)GetStream()andDoExchange()correctly decompress whenICompressionCodecFactoryis providedICompressionCodecFactoryisnull(existing behavior)